package com.lzy.canal.service;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.lzy.canal.component.filter.mysql.MySqlFilter;
import com.lzy.canal.component.result.BaseConsumerResultData;
import com.lzy.canal.model.CanalMessageEntryResultData;
import com.lzy.canal.register.consumer.MysqlFilterRegister;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * Canal 消息服务
 *
 * @author lzy
 */
@Slf4j
public class CanalMessageService {

    private final CanalConsumerService canalConsumerService;

    private final MysqlFilterRegister filterRegister;


    public CanalMessageService(CanalConsumerService canalConsumerService, MysqlFilterRegister filterRegister) {
        this.canalConsumerService = canalConsumerService;
        this.filterRegister = filterRegister;

    }

    public List<CanalMessageEntryResultData> consumerMessage(Message message) {
        return message.getEntries().stream()
                .filter(o -> !CanalEntry.EntryType.TRANSACTIONBEGIN.equals(o.getEntryType()) && !CanalEntry.EntryType.TRANSACTIONEND.equals(o.getEntryType()))
                .map(o -> {
                    CanalEntry.Header header = o.getHeader();
                    MySqlFilter filter = filterRegister.getFilter(header.getSchemaName());
                    if (!filter.apply( header.getTableName(),header.getSchemaName())) {
                        log.info("{}-->{} 未配置", header.getSchemaName(), header.getTableName());
                        log.info("type-->{}", o.getEntryType());
                        return null;
                    }
                    try {
                        List<? extends BaseConsumerResultData> consumerResultData = canalConsumerService.consumer(o);
                        return CanalMessageEntryResultData.builder()
                                .results(consumerResultData.toArray(new BaseConsumerResultData[0]))
                                .tableName(header.getTableName())
                                .schemaName(header.getSchemaName())
                                .build();
                    } catch (Exception e) {
                        log.error("解析mysql日志消息失败", e);
                        return null;
                    }
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

}
